/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <errno.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>

#include <cstdint>
#include <limits>
#include <string>
#include <sstream>
#include <limits>

#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/common/file_wrapper.h>

namespace Hdfs {
    namespace Internal {
        MappedFileWrapper::MappedFileWrapper() : delegate(true), begin(nullptr), position(nullptr), fd(-1), size(0) {
        }

        MappedFileWrapper::~MappedFileWrapper() {
            close();
        }

        bool MappedFileWrapper::openInternal(int fd, bool delegate, size_t size) {
            this->delegate = delegate;
#ifdef MAP_NOCACHE
            void *retval = mmap(nullptr, size, PROT_READ, MAP_FILE | MAP_PRIVATE | MAP_NOCACHE, fd, 0);
#else
            void *retval = mmap(nullptr, size, PROT_READ, MAP_FILE | MAP_PRIVATE, fd, 0);
#endif
            begin = position = static_cast<const char *>(retval);

            if (MAP_FAILED == retval) {
                begin = position = nullptr;
                close();
                return false;
            }

            if (posix_madvise(const_cast<char *>(begin), size, POSIX_MADV_SEQUENTIAL)) {
                close();
                return false;
            }

            return true;
        }

        bool MappedFileWrapper::open(int fd, bool delegate) {
            size = lseek(fd, 0, SEEK_END);
            lseek(fd, 0, SEEK_SET);
            std::stringstream ss;
            ss.imbue(std::locale::classic());
            ss << "FileDescriptor " << fd;
            path = ss.str();

            if (static_cast<uint64_t>(size) > static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
                THROW(HdfsIOException,
                      "Cannot create memory mapped file for \"%s\", file is too large.",
                      path.c_str());
            }

            return openInternal(fd, delegate, static_cast<size_t>(size));
        }

        bool MappedFileWrapper::open(const std::string &path) {
            struct stat st;

            if (stat(path.c_str(), &st)) {
                return false;
            }

            size = st.st_size;

            if (static_cast<uint64_t>(size) > static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
                THROW(HdfsIOException,
                      "Cannot create memory mapped file for \"%s\", file is too large.",
                      path.c_str());
            }

            fd = ::open(path.c_str(), O_RDONLY);

            if (fd < 0) {
                return false;
            }

            this->path = path;
            return openInternal(fd, true, st.st_size);
        }

        void MappedFileWrapper::close() {
            if (nullptr != begin) {
                ::munmap(const_cast<char *>(begin), static_cast<size_t>(size));
                begin = position = nullptr;
            }

            if (fd >= 0 && delegate) {
                ::close(fd);
            }

            fd = -1;
            size = 0;
            delegate = true;
            path.clear();
        }

        const char *MappedFileWrapper::read(std::vector<char> &buffer, int32_t size) {
            assert(nullptr != begin && nullptr != position);
            const char *retval = position;
            position += size;
            return retval;
        }

        void MappedFileWrapper::copy(char *buffer, int32_t size) {
            assert(nullptr != begin && nullptr != position);
            memcpy(buffer, position, size);
            position += size;
        }

        void MappedFileWrapper::seek(int64_t offset) {
            assert(nullptr != begin && nullptr != position);
            position = begin + offset;
        }
    }
}
